package com.atguigu.userprofile.app;

import com.atguigu.userprofile.bean.TagInfo;
import com.atguigu.userprofile.constants.ConstCodes;
import com.atguigu.userprofile.dao.TagInfoDAO;
import com.atguigu.userprofile.util.MyClickhouseUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

public class TaskBitmapApp {


    //1 需要启用的标签集合
    //
    //
    //2 建表
    //每天建(自动建)   永久建(手动建)
    //永久建立4张表  根据标签值的类型
    // 手动执行建表语句
    //
    //
    //3 读    insert  select
    //   写
    public static void main(String[] args) {

        // 0 该程序本身不是分布式程序，  创建spark上下文，把该程序作为spark程序的注意目的，和其他程序一样便于统一调度管理。
        SparkConf sparkConf = new SparkConf().setAppName("task_bitmap_app");
        SparkContext sparkContext = new SparkContext(sparkConf);

        String busiDate = args[1];

        //1 需要启用的标签集合
        List<TagInfo> tagInfoList = TagInfoDAO.getTagInfoListWithOn();

        //2 建表手动

        //3  因为不同标签值类型的标签要进入不同的bitmap表
        // 要根据标签值类型把全部标签集合分成4份

        List<TagInfo> tagInfoListString =new ArrayList<>();
        List<TagInfo> tagInfoListLong =new ArrayList<>();
        List<TagInfo> tagInfoListDecimal =new ArrayList<>();
        List<TagInfo> tagInfoListDate =new ArrayList<>();

        for (TagInfo tagInfo : tagInfoList) {
            if(tagInfo.getTagValueType().equals(ConstCodes.TAG_VALUE_TYPE_STRING)){
                tagInfoListString.add(tagInfo);
            }else if(tagInfo.getTagValueType().equals(ConstCodes.TAG_VALUE_TYPE_LONG)){
                tagInfoListLong.add(tagInfo);
            } else if(tagInfo.getTagValueType().equals(ConstCodes.TAG_VALUE_TYPE_DECIMAL)){
                tagInfoListDecimal.add(tagInfo);
            }  else   if(tagInfo.getTagValueType().equals(ConstCodes.TAG_VALUE_TYPE_DATE)){
                tagInfoListDate.add(tagInfo);
            }
        }

        //要根据标签值类型分别组成sql
        insertBitmap(tagInfoListString,ConstCodes.TAG_VALUE_TYPE_STRING,"user_tag_value_string", busiDate);
        insertBitmap(tagInfoListLong,ConstCodes.TAG_VALUE_TYPE_LONG,"user_tag_value_long", busiDate);
        insertBitmap(tagInfoListDecimal,ConstCodes.TAG_VALUE_TYPE_DECIMAL,"user_tag_value_decimal", busiDate);
        insertBitmap(tagInfoListDate,ConstCodes.TAG_VALUE_TYPE_DATE,"user_tag_value_date", busiDate);

    }

    private static void  insertBitmap(List<TagInfo> tagInfoList,String tagValueType ,String targetTableName,String busiDate){

        if(tagInfoList.size()==0){
            return;
        }

        //清理当日数据  //幂等性
        //alter table $targetTableName  delete where dt='$busiDate'

        String deleteSQL="alter table  "+targetTableName+" delete where dt='"+busiDate+"'";
        System.out.println(deleteSQL);
        MyClickhouseUtil.executeSql(deleteSQL);

        //来源表名
        String sourceTable="user_tag_merge_"+busiDate.replace("-","");
        List<String> tagCodeList = tagInfoList.stream().map(tagInfo -> "('" + tagInfo.getTagCode().toLowerCase() + "'," + tagInfo.getTagCode().toLowerCase() + ")").collect(Collectors.toList());

        String tagCodeSQL = StringUtils.join(tagCodeList, ",");


        //insert into user_tag_value_string
        //select  tv.1 as tag_code,tv.2  as tag_value, groupBitmapState( cast(uid as UInt64)  )  ,'2020-06-14' from
        //(
        //select uid, arrayJoin([('tg_person_base_gender',tg_person_base_gender),('tg_person_base_agegroup',tg_person_base_agegroup)] ) tv
        //from user_tag_merge_20200614 ) tt
        //group by tv.1 ,tv.2
        String tv2="tv.2";
        if(tagValueType.equals(ConstCodes.TAG_VALUE_TYPE_LONG)||tagValueType.equals(ConstCodes.TAG_VALUE_TYPE_DECIMAL)){
              tv2=" if(tv.2 ='' ,'0',tv.2)" ;
        }

        String insertSQL="   insert into " +targetTableName+  "\n"+
                "       select  tv.1 as tag_code, "+tv2+" as tag_value, groupBitmapState( cast(uid as UInt64)  )  ,'"+busiDate+"' from\n" +
                "         (\n" +
                "       select uid, arrayJoin([ "+tagCodeSQL+"] ) tv\n" +
                "        from  "+sourceTable+" ) tt\n" +
                "         group by tv.1 ,tv.2";
        System.out.println(insertSQL);

        MyClickhouseUtil.executeSql(insertSQL);
    }
}
